package com.lobstar.xrpc.client.main;

import java.io.Closeable;
import java.util.ArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.lobstar.xrpc.client.handler.ClientServiceHandler;
import com.lobstar.xrpc.client.handler.HeartBeatHandler;
import com.lobstar.xrpc.common.ClientCallable;
import com.lobstar.xrpc.common.TaskExecutable;
import com.lobstar.xrpc.common.handler.SMessageDecoder;
import com.lobstar.xrpc.common.handler.SMessageEncoder;
import com.lobstar.xrpc.common.po.Const;
import com.lobstar.xrpc.common.po.SMessage;
import com.lobstar.xrpc.common.po.Task;
import com.lobstar.xrpc.context.ClientContainer;
import com.lobstar.xrpc.server.handler.task.TaskCallback;
import com.lobstar.xrpc.server.listener.OperationListener;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;

public class MainClient extends TaskExecutable implements ClientCallable,Closeable{
	private static final Logger LOGGER = LoggerFactory.getLogger(MainClient.class);
	private static final int MAX_LENGTH = 5242880;
	//private static Map<String, MainClient> clientMap = new ConcurrentHashMap<String, MainClient>();
	private ExecutorService clientRunner = Executors.newFixedThreadPool(1);

	private String ip = "10.131.1.42";
	private int port = 10022;
	private String clientId;

	private EventLoopGroup group;
	private Channel channel;

	private OperationListener operationListener;
	private String basePackage;

	private int reconnectInteval = 10;
	private boolean reconnect = true;
	private boolean online = false;

	//private BlockingQueue<Task> taskQueue = new ArrayBlockingQueue<Task>(500);;

	private MainClient(String id, String ip, int port,String basePackage) {
		super(basePackage,10,100,10);
		this.clientId = id;
		this.ip = ip;
		this.port = port;
		this.basePackage = basePackage;
	}

	private MainClient(String id, String ip, int port,String basePackage,int taskExecutorSize,int responseCacheSize,int responseCacheExpire) {
		super(basePackage,taskExecutorSize,responseCacheSize,responseCacheExpire);
		this.clientId = id;
		this.ip = ip;
		this.port = port;
		this.basePackage = basePackage;
	}


	public static MainClient build(String id, String ip, int port,String basePackage) throws InterruptedException {
		if(ClientContainer.exist(id)) {
			throw new RuntimeException("client id repeat !");
		}
		MainClient client = new MainClient(id, ip, port,basePackage);
		ClientContainer.addClient(id, client);
		return client;
	}

	public static MainClient get(String id) {
		return ClientContainer.getClient(id);
	}

	public void start() throws InterruptedException {
		clientRunner.execute(new Runnable() {
			@Override
			public void run() {
				connect();
			}
		});
	}

	public MainClient addOperationListener(OperationListener operationListener) {
		this.operationListener = operationListener;
		return this;
	}

	private String sendMessage(SMessage message) {
		if (message == null) {
			throw new RuntimeException("cannot sent null message");
		}
		ChannelFuture future = this.channel.writeAndFlush(message);
		future.addListener(new GenericFutureListener<Future<? super Void>>() {
			@Override
			public void operationComplete(Future<? super Void> future) throws Exception {
				if (future.isSuccess()) {
					// sendListener.success(message);
				} else {
					// sendListener.fail(message);
				}
			}
		});
		return message.getId();
	}

	public void call(SMessage message, TaskCallback callback) {
		Task task = new Task();
		task.setRequest(message);
		task.setCallback(callback);
		task.setId(message.getId());
		String id = task.getId();
		cacheTask(id,task);
		sendMessage(task.getRequest());
	}

	private void connect() {
		group = new NioEventLoopGroup(10);
		try {
			Bootstrap bootstrap = new Bootstrap().group(group).channel(NioSocketChannel.class)
					.handler(this.new SimpleClientInitializer());
			channel = bootstrap.remoteAddress(ip, port).connect()
					.addListener(new GenericFutureListener<Future<? super Void>>() {
						@Override
						public void operationComplete(Future<? super Void> future) throws Exception {
							// TODO Auto-generated method stub
							if (future.isSuccess()) {
								online = true;
								LOGGER.info("与服务器{}:{}连接建立成功...", ip, port);
								if(operationListener!= null) {
									operationListener.success();
								}
							} else {
								online = false;
								LOGGER.info("与服务器{}:{}连接建立失败...", ip, port);
								if(operationListener != null) {
									operationListener.fail();
								}
							}
						}
					}).channel();
			channel.closeFuture().sync();
		} catch (Exception e) {
			LOGGER.error(e.getMessage(), e);
		} finally {
			if(reconnect) {
				group.shutdownGracefully();
				clientRunner.execute(new Runnable() {
					@Override
					public void run() {
						reConnect();
					}
				});
			}
		}
	}

	class SimpleClientInitializer extends ChannelInitializer<SocketChannel> {
		@Override
		public void initChannel(SocketChannel ch) throws Exception {
			ChannelPipeline pipeline = ch.pipeline();
			pipeline.addLast(new DelimiterBasedFrameDecoder(MAX_LENGTH, Unpooled.copiedBuffer(Const.DELIMITER.getBytes("UTF-8"))));
			pipeline.addLast(new SMessageEncoder());
			pipeline.addLast(new SMessageDecoder());
			pipeline.addLast(new IdleStateHandler(0, 0, 60, TimeUnit.SECONDS));
			pipeline.addLast(new HeartBeatHandler(clientId));
			pipeline.addLast(new ClientServiceHandler(MainClient.this));
		}
	}

	private void reConnect() {
		// fixme: 重连显式退出?
		try {
			LOGGER.info("与服务器{}:{}连接已断开, {}秒后重连...", ip, port, reconnectInteval);
			online = false;
			Thread.sleep(reconnectInteval * 1000);
			connect();
		} catch (Exception e) {
			LOGGER.error(e.getMessage(), e);
		}
	}

	public MainClient blockStart() throws InterruptedException {
		start();
		int wait = 100;
		while(!online) {
			--wait;
			if(wait==0) {break;}
			Thread.sleep(100);
		}
		if(!online) {
			throw new RuntimeException("connot connect server");
		}
		return this;
	}

	public void checkOnline() {
		if(!online) {
			throw new RuntimeException("not online");
		}
	}

	public void close() {
		reconnect = false;
		channel.close();
		group.shutdownGracefully();
		clientRunner.shutdown();
	}

	public void setReconnectInteval(int reconnectInteval) {
		this.reconnectInteval = reconnectInteval;
	}

	interface TransformObject {
		public void handle(ArrayList<?> ret,String json);
	}

	@Override
	public boolean online() {
		return online;
	}

}
